package com.源码分析;

import org.junit.Ignore;
import org.junit.Test;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.CoreSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class Demo1 {

	private static final Logger log = LoggerFactory.getLogger(Demo1.class);

	@Test
	@Ignore
	public void test1() {
		Flux.just(1, 2, 3, 4, 5).subscribe(new CoreSubscriber<Integer>() {// 这里传入CoreSubscriber对象作为订阅者
			private Subscription subscription;
			@Override
			public void onSubscribe(Subscription s) {
				// 保存订阅关系, 需要用它来给发布者响应
                this.subscription = s;
				log.info("onSubscribe, {}", s.getClass());
				// 请求5个数据
				s.request(5);
			}

			@Override
			public void onNext(Integer integer) {
				log.info("onNext： {}", integer);
				// 处理完调用request再请求一个数据
                this.subscription.request(1);
                // 或者 已经达到了目标, 调用cancel告诉发布者不再接受数据了
                // this.subscription.cancel();
			}

			@Override
			public void onError(Throwable t) {
				// 出现了异常(例如处理数据的时候产生了异常)
                t.printStackTrace();

				// 我们可以告诉发布者, 后面不接受数据了
                this.subscription.cancel();
			}

			@Override
			public void onComplete() {
				// 全部数据处理完了(发布者关闭了)
				log.info("onComplete");
			}
		});
	}

	@Test
	@Ignore
	public void test2() {
		Flux.just(1, 2, 3, 4, 5)
		.map(x -> x * 2) // map这种即是订阅者又充当发布者的角色
		.subscribe(new CoreSubscriber<Integer>() {// 这里传入CoreSubscriber对象作为订阅者
			@Override
			public void onSubscribe(Subscription s) {
				log.info("onSubscribe, {}", s.getClass());
				s.request(5);
			}

			@Override
			public void onNext(Integer integer) {
				log.info("onNext： {}", integer);
			}

			@Override
			public void onError(Throwable t) {
			}

			@Override
			public void onComplete() {
				log.info("onComplete");
			}
		});
	}

	@Test
	@Ignore
	public void test3() {
		Flux.just("flux", "mono")
		.map(x -> {
			return x + "-ok";
		})
		.doOnNext(x -> {
			System.out.println(x);
		})
		.flatMap(x -> {
			return Flux.fromArray(x.split("\\s*"));
		})
		.filter(x -> {
			return !x.equals("o");
		})
		.subscribe(x -> {
			System.out.println(x);
		});
	}

	@Test
	public void test4() {
		Flux.just("flux", "mono")
		.map(x -> {
			System.out.println("map: " + Thread.currentThread().getName());
			return x + "-ok";
		})
		.filter(x -> {
			System.out.println("filter: " + Thread.currentThread().getName());
			return !x.equals("o");
		})
		.publishOn(Schedulers.newElastic("publishOn"))
		.map(x -> {
			System.out.println("map: " + Thread.currentThread().getName());
			return x + "-no";
		})
		.subscribe(x -> {
			System.out.println("subscribe: " + Thread.currentThread().getName());
			System.out.println(x);
		});

		try {
			Thread.sleep(100000000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Test
	@Ignore
	public void test5() {
		Flux.just("flux", "mono")
		.map(x -> {
			System.out.println("map: " + Thread.currentThread().getName());
			return x + "-ok";
		})
		.doOnNext(x -> {
			System.out.println("doOnNext: " + Thread.currentThread().getName());
			System.out.println(x);
		})
		.flatMap(x -> {
			System.out.println("flatMap: " + Thread.currentThread().getName());
			return Flux.fromArray(x.split("\\s*"));
		})
		.subscribeOn(Schedulers.elastic())
		.publishOn(Schedulers.newElastic("publishOn"))
		.filter(x -> {
			System.out.println("filter: " + Thread.currentThread().getName());
			return !x.equals("o");
		})
		.subscribe(x -> {
			System.out.println("subscribe: " + Thread.currentThread().getName());
			System.out.println(x);
		});

		try {
			Thread.sleep(100000000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}
